﻿using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Polly;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System.Net.Sockets;
using System.Text;
using System.Text.Json;
using XYS.Core.Models;
using XYS.Core.Util;
using XYS.Service.Core.EventBus.Abstractions;
using XYS.Service.Core.EventBus.Events;
using XYS.Service.Core.EventBus.Extensions;
using XYS.Service.Core.Models;
using XYS.Service.Core.Util.Converters;
using XYS.Service.Core.Util.Helper;

namespace XYS.Service.Core.EventBus.EventBusRabbitMQ
{
    /// <summary>
    /// 使用RabbitMQ的事件总线
    /// </summary>
    public class EventBusRabbitMQ : IEventBus, IDisposable
    {
        private readonly IRabbitMQPersistentConnection persistentConnection;
        private readonly ILogger<EventBusRabbitMQ> logger;
        private readonly IEventBusSubscriptionsManager subsManager;
        private readonly int retryCount;

        private IModel consumerChannel;
        private string queueName;
        private readonly IServiceProvider serviceProvider;
        private Nest.IElasticClient esClient;

        /// <summary>
        /// 使用RabbitMQ的事件总线
        /// </summary>
        public EventBusRabbitMQ(IServiceProvider serviceProvider, IRabbitMQPersistentConnection persistentConnection, ILogger<EventBusRabbitMQ> logger,
             IEventBusSubscriptionsManager subsManager, string queueName = null, int retryCount = 5)
        {
            this.persistentConnection = persistentConnection ?? throw new ArgumentNullException(nameof(persistentConnection));
            this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
            this.subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
            this.queueName = queueName;
            consumerChannel = CreateConsumerChannel();
            this.retryCount = retryCount;
            this.subsManager.OnEventRemoved += SubsManager_OnEventRemoved;

            this.serviceProvider = serviceProvider;
            esClient = ElasticClientHelper.GetClient(GlobalCoreSetting.EventBugSetting.ElasticName);
        }


        /// <summary>
        /// 订阅管理器删除事件
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="eventName"></param>
        private void SubsManager_OnEventRemoved(object sender, string eventName)
        {
            if (!persistentConnection.IsConnected)
            {
                persistentConnection.TryConnect();
            }

            using (var channel = persistentConnection.CreateModel())
            {
                channel.QueueUnbind(queue: queueName,
                    exchange: GlobalCoreSetting.EventBugSetting.ExchangeName,
                    routingKey: eventName);

                if (subsManager.IsEmpty)
                {
                    queueName = string.Empty;
                    consumerChannel.Close();
                }
            }
        }

        /// <summary>
        /// 发布
        /// </summary>
        /// <param name="event"></param>
        public void Publish(IntegrationEvent @event)
        {
            if (!persistentConnection.IsConnected)
            {
                persistentConnection.TryConnect();
            }

            var policy = Policy.Handle<BrokerUnreachableException>()
                .Or<SocketException>()
                .WaitAndRetry(retryCount, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (ex, time) =>
                {
                    logger.LogWarning(ex, "Could not publish event: {EventId} after {Timeout}s ({ExceptionMessage})", @event.Id, $"{time.TotalSeconds:n1}", ex.Message);
                });

            var eventName = @event.GetType().Name;

            logger.LogTrace("Creating RabbitMQ channel to publish event: {EventId} ({EventName})", @event.Id, eventName);

            using (var channel = persistentConnection.CreateModel())
            {
                logger.LogTrace("Declaring RabbitMQ exchange to publish event: {EventId}", @event.Id);

                channel.ExchangeDeclare(exchange: GlobalCoreSetting.EventBugSetting.ExchangeName, type: "direct");

                var body = JsonSerializer.SerializeToUtf8Bytes(@event, @event.GetType(), new JsonSerializerOptions
                {
                    WriteIndented = true
                });

                policy.Execute(() =>
                {
                    var properties = channel.CreateBasicProperties();
                    properties.DeliveryMode = 2; // persistent

                    logger.LogTrace("Publishing event to RabbitMQ: {EventId}", @event.Id);

                    channel.BasicPublish(
                        exchange: GlobalCoreSetting.EventBugSetting.ExchangeName,
                        routingKey: eventName,
                        mandatory: true,
                        basicProperties: properties,
                        body: body);


                    //记录消费日志
                    if (esClient != null && GlobalCoreSetting.EventBugSetting.LogPublishEventNameList.Contains(eventName))
                    {
                        try
                        {
                            EventLogModel logEntity = new()
                            {
                                EventName = eventName,
                                Message = @event.ToJson(),
                                TriggerTime = DateTime.Now,
                            };

                            //保存记录至ES
                            string indexName = $"{GlobalCoreSetting.EventBugSetting.ElasticNamePublish}-{GlobalCoreSetting.ServiceName}-{GlobalCoreSetting.EnvironmentName}-{DateTime.Now:yyyy-MM-dd}".ToLower();
                            Task.Run(() =>
                            {
                                var resultBulkES = esClient.Bulk(indexName, logEntity);
                                if (!resultBulkES.Succeeded) logger.LogError($"【事件总线-保存日志】发生异常.{resultBulkES.ToJson()}");
                            }).DoNotAwait();
                        }
                        catch (Exception ex) { logger.LogError($"【事件总线-保存日志】发生异常.{ex.ToJson()}"); }
                    }

                });
            }
        }

        /// <summary>
        /// 动态订阅
        /// </summary>
        /// <typeparam name="TH"></typeparam>
        /// <param name="eventName"></param>
        public void SubscribeDynamic<TH>(string eventName)
            where TH : IDynamicIntegrationEventHandler
        {
            logger.LogInformation("Subscribing to dynamic event {EventName} with {EventHandler}", eventName, typeof(TH).GetGenericTypeName());

            DoInternalSubscription(eventName);
            subsManager.AddDynamicSubscription<TH>(eventName);
            StartBasicConsume();
        }

        /// <summary>
        /// 订阅
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <typeparam name="TH"></typeparam>
        public void Subscribe<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventHandler<T>
        {
            var eventName = subsManager.GetEventKey<T>();
            DoInternalSubscription(eventName);

            logger.LogInformation("Subscribing to event {EventName} with {EventHandler}", eventName, typeof(TH).GetGenericTypeName());

            subsManager.AddSubscription<T, TH>();
            StartBasicConsume();
        }

        /// <summary>
        /// 做内部订阅
        /// </summary>
        /// <param name="eventName"></param>
        private void DoInternalSubscription(string eventName)
        {
            var containsKey = subsManager.HasSubscriptionsForEvent(eventName);
            if (!containsKey)
            {
                if (!persistentConnection.IsConnected)
                {
                    persistentConnection.TryConnect();
                }

                consumerChannel.QueueBind(queue: queueName,
                                    exchange: GlobalCoreSetting.EventBugSetting.ExchangeName,
                                    routingKey: eventName);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        public void Unsubscribe<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventHandler<T>
        {
            var eventName = subsManager.GetEventKey<T>();

            logger.LogInformation("Unsubscribing from event {EventName}", eventName);

            subsManager.RemoveSubscription<T, TH>();
        }

        /// <summary>
        /// 
        /// </summary>
        public void UnsubscribeDynamic<TH>(string eventName)
            where TH : IDynamicIntegrationEventHandler
        {
            subsManager.RemoveDynamicSubscription<TH>(eventName);
        }

        /// <summary>
        /// 释放
        /// </summary>
        public void Dispose()
        {
            if (consumerChannel != null)
            {
                consumerChannel.Dispose();
            }

            subsManager.Clear();
        }

        /// <summary>
        /// 开始基本消费
        /// </summary>
        private void StartBasicConsume()
        {
            logger.LogTrace("Starting RabbitMQ basic consume");

            if (consumerChannel != null)
            {
                var consumer = new AsyncEventingBasicConsumer(consumerChannel);

                consumer.Received += Consumer_Received;

                consumerChannel.BasicConsume(
                    queue: queueName,
                    autoAck: false,
                    consumer: consumer);
            }
            else
            {
                logger.LogError("StartBasicConsume can't call on _consumerChannel == null");
            }
        }

        /// <summary>
        /// 消费者收到消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="eventArgs"></param>
        /// <returns></returns>
        private async Task Consumer_Received(object sender, BasicDeliverEventArgs eventArgs)
        {
            var eventName = eventArgs.RoutingKey;
            var message = Encoding.UTF8.GetString(eventArgs.Body.Span);

            try
            {
                if (message.ToLowerInvariant().Contains("throw-fake-exception"))
                    throw new InvalidOperationException($"Fake exception requested: \"{message}\"");

                if (GlobalCoreSetting.EventBugSetting.BlackEventNameList.Contains(eventName) || !subsManager.HasSubscriptionsForEvent(eventName) || !subsManager.GetHandlersForEvent(eventName).Any())
                {
                    logger.LogError($"时间总线未获取到相关处理器【{eventName}】");
                    consumerChannel.BasicAck(eventArgs.DeliveryTag, multiple: false);
                    return;
                }

                //记录消费日志
                if (esClient != null && GlobalCoreSetting.EventBugSetting.LogEventNameList.Contains(eventName))
                {
                    Task.Run(() =>
                    {
                        try
                        {
                            var eventLogModel = new EventLogModel
                            {
                                EventName = eventName,
                                Message = message,
                                TriggerTime = DateTime.Now,
                            };
                            string indexName = $"{GlobalCoreSetting.EventBugSetting.ElasticName}-{GlobalCoreSetting.ServiceName}-{GlobalCoreSetting.EnvironmentName}-{DateTime.Now:yyyy-MM-dd}".ToLower();
                            //保存记录至ES
                            var resultBulkES = esClient.Bulk(indexName, eventLogModel);
                            if (!resultBulkES.Succeeded) logger.LogError($"【事件总线-保存日志】发生异常：{resultBulkES.ToJson()}\r\nIndexName:{indexName}\r\n请求参数:{eventLogModel.ToJson()}");
                        }
                        catch (Exception ex) { logger.LogError($"【事件总线-保存日志】发生异常.{ex.ToJson()}"); }
                    }).DoNotAwait();
                }

                await ProcessEvent(eventName, message);
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "----- ERROR Processing message \"{Message}\"", message);
            }
            consumerChannel.BasicAck(eventArgs.DeliveryTag, multiple: false);
        }

        private IModel CreateConsumerChannel()
        {
            if (!persistentConnection.IsConnected)
            {
                persistentConnection.TryConnect();
            }

            logger.LogTrace("Creating RabbitMQ consumer channel");

            var channel = persistentConnection.CreateModel();

            channel.ExchangeDeclare(exchange: GlobalCoreSetting.EventBugSetting.ExchangeName,
                                    type: "direct");

            channel.QueueDeclare(queue: queueName,
                                 durable: true,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: null);

            channel.CallbackException += (sender, ea) =>
            {
                logger.LogWarning(ea.Exception, "Recreating RabbitMQ consumer channel");

                consumerChannel.Dispose();
                consumerChannel = CreateConsumerChannel();
                StartBasicConsume();
            };

            return channel;
        }

        /// <summary>
        /// 进程事件（使用autofac）推荐
        /// </summary>
        /// <param name="eventName"></param>
        /// <param name="message"></param>
        /// <returns></returns>
        private async Task ProcessEvent(string eventName, string message)
        {
            logger.LogTrace("Processing RabbitMQ event: {EventName}", eventName);

            if (subsManager.HasSubscriptionsForEvent(eventName))
            {
                var subscriptions = subsManager.GetHandlersForEvent(eventName);
                if (!subscriptions.Any()) return;

                using var scope = serviceProvider.CreateAsyncScope();
                foreach (var subscription in subscriptions)
                {
                    if (subscription.IsDynamic)
                    {
                        if (scope.ServiceProvider.GetService(subscription.HandlerType) is not IDynamicIntegrationEventHandler handler) continue;
                        using dynamic eventData = JsonDocument.Parse(message);
                        await Task.Yield();
                        await handler.Handle(eventData);
                    }
                    else
                    {
                        if (scope.ServiceProvider.GetService(subscription.HandlerType) == null) continue;
                        var eventType = subsManager.GetEventTypeByName(eventName);
                        var options = new JsonSerializerOptions
                        {
                            PropertyNameCaseInsensitive = true,
                            Converters = { new CustomDateTimeConverter() }
                        };
                        //var integrationEvent = JsonSerializer.Deserialize(message, eventType, options);
                        var integrationEvent = message.ToObject(eventType);
                        var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);

                        await Task.Yield();
                        await (Task)concreteType.GetMethod("Handle").Invoke(scope.ServiceProvider.GetService(subscription.HandlerType), new object[] { integrationEvent });
                    }
                }
            }
            else
            {
                logger.LogWarning("No subscription for RabbitMQ event: {EventName}", eventName);
            }
        }

        /// <summary>
        /// 进程事件（使用自带的）
        /// </summary>
        /// <param name="eventName"></param>
        /// <param name="message"></param>
        /// <returns></returns>
        private async Task ProcessEventByNetCore(string eventName, string message)
        {
            logger.LogTrace("Processing RabbitMQ event: {EventName}", eventName);

            if (subsManager.HasSubscriptionsForEvent(eventName))
            {
                //安装 Microsoft.Extensions.DependencyInjection扩展包

                using (var scope = serviceProvider.CreateScope())
                {
                    var subscriptions = subsManager.GetHandlersForEvent(eventName);
                    foreach (var subscription in subscriptions)
                    {
                        if (subscription.IsDynamic)
                        {
                            var handler = scope.ServiceProvider.GetRequiredService(subscription.HandlerType) as IDynamicIntegrationEventHandler;
                            if (handler == null) continue;
                            using dynamic eventData = JsonDocument.Parse(message);
                            await Task.Yield();
                            await handler.Handle(eventData);
                        }
                        else
                        {
                            var handler = scope.ServiceProvider.GetRequiredService(subscription.HandlerType);
                            if (handler == null) continue;
                            var eventType = subsManager.GetEventTypeByName(eventName);
                            var integrationEvent = JsonSerializer.Deserialize(message, eventType, new JsonSerializerOptions() { PropertyNameCaseInsensitive = true });
                            var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType);

                            await Task.Yield();
                            await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent });
                        }
                    }
                }




            }
            else
            {
                logger.LogWarning("No subscription for RabbitMQ event: {EventName}", eventName);
            }
        }

    }
}
